package pageRank;

import java.io.IOException;

import mapred.AbstractMapper;
import mapred.util.Bytes;

import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

import removeDanglingPage.ReadBytes;

public class GetReady_Mapper extends AbstractMapper<BytesWritable, BytesWritable, BytesWritable, BytesWritable> {

	@Override
	public void map(BytesWritable key, BytesWritable value,
			OutputCollector<BytesWritable, BytesWritable> output,
			Reporter reporter) throws IOException {
		// TODO Auto-generated method stub
		if(value.getLength() > 0) {
			ReadBytes v = new ReadBytes(value);
			byte[] link = Bytes.concat(Bytes.Double2Bytes(1d), v.nextBytes(v.size()));
			output.collect(key, new BytesWritable(link));
		} else {
			output.collect(key, new BytesWritable(Bytes.Double2Bytes(1d)));
		}
//		if(value.getLength() > 0) {
//			ReadBytes v = new ReadBytes(value);
//			output.collect(key, value);
//			output.collect(key, new BytesWritable(Bytes.Double2Bytes(-1d)));
//			while(v.hasNextUrl()) {
//				output.collect(new BytesWritable(v.nextUrl()), new BytesWritable(Bytes.Double2Bytes(-1d)));
//			}
//		}
	}

//	@Override
//	public void map(Text key, Text value, OutputCollector<Text, Text> output,
//			Reporter reporter) throws IOException {
//		output.collect(key, value);
//		output.collect(key, new Text("#-1"));
//		if (!value.toString().trim().isEmpty()) {
//			String[] targets = value.toString().split("\t");
//			for (String target : targets) {
//				output.collect(new Text(target), new Text("#-1"));
//			}
//		}
//	}

}
